package com.spider.moudle.task;

import com.spider.moudle.constant.Constant;
import com.spider.moudle.entity.Task;
import com.spider.moudle.factory.KafkaConsumerFactory;

import kafka.consumer.ConsumerIterator;
import kafka.message.MessageAndMetadata;

/**
 * Created by Linbo Hu on 2017/11/29.
 * 任务调度器
 */
public class TaskWorker implements Runnable {


    private ConsumerIterator<String, String> consumer;


    private String topic;

    public TaskWorker(String topic) {
        this.topic = topic;
        consumer = KafkaConsumerFactory.build(topic);
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            while (consumer.hasNext()) {
                if (TaskManager.isOverLoad(topic)) {
                    try {
                        Thread.sleep(Constant.TASK_POLL_ITL);
                        continue;
                    } catch (InterruptedException e) {
                        break;
                    }
                }
                MessageAndMetadata<String, String> record = consumer.next();
                String taskName = record.key();
                Task task = TaskManager.parse(record.message());
                try {
                    TaskManager.addLocal(taskName, task);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 清除topic中的数据
     */
    public void clear() {
        long i=0;
            while (consumer.hasNext()) {
                i++;
                MessageAndMetadata<String, String> record = consumer.next();
                Task task = TaskManager.parse(record.message());
                System.out.println("清理第："+i+"条数据");
            }
    }


    public static void main(String[] args) {
        //1101_SPIDER_SPEECH || 1101_SPIDER_USER
        String topic = "1101_SPIDER_SPEECH";
        new TaskWorker(topic).clear();
    }
}
